package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/18 23:10:13
 * @version: 1.0
 */
public class AdjustIsNewProcessFuncion extends KeyedProcessFunction<String, String, String> {

    private transient ValueState<String> firstDateState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
//        2.初始化状态
        firstDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("first_Date", String.class));
        super.open(parameters);
    }

    @Override
 /**
 * 处理元素函数，根据输入值更新元素的新旧状态
 *
 * @param value 输入的JSON字符串，包含需要处理的数据
 * @param ctx 上下文对象，用于获取状态信息
 * @param out 用于输出处理后的数据
 * @throws Exception 如果处理过程中发生错误，则抛出异常
 */
public void processElement(String value, Context ctx, Collector<String> out) throws Exception {

    // 获取首次出现日期的状态值
    String firstDate = firstDateState.value();

    // 解析输入的JSON字符串
    JSONObject jsonObject = JSON.parseObject(value);

    // 获取元素是否为新的标志
    String isNewValue = jsonObject.getJSONObject("common").getString("is_new");

    // 获取时间戳值
    Long tsValue = jsonObject.getLong("ts");

    // 如果元素标记为新
    if ("1".equals(isNewValue)) {

        // 如果首次出现日期状态为空
        if (null == firstDate) {
            // 更新首次出现日期状态为当前时间戳对应的日期
            firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
        } else {

            // 如果首次出现日期与当前时间戳对应的日期不相同
            if (!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))){
                // 更新元素为旧的状态
                jsonObject.getJSONObject("common").put("is_new", "0");
            }
        }
    } else {
//         如果元素不为新  且首次出现日期状态为空

//         当用户早早登陆 ，但是 实时应用后来处理，用户第一次访问的时 firstDate 为空
        if (null == firstDate){
            // 更新首次出现日期状态为当前时间戳减去一天对应的日期
            firstDateState.update(DateTimeUtil.convertLongToString(tsValue-86400000, DateTimeUtil.DATE_FORMAT));
        }
    }

    // 输出处理后的nJSON字符串
    out.collect(jsonObject.toJSONString());
}

}

